/*udt server*/
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <netdb.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <string.h>
#include <errno.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <error.h>
#include <signal.h>
#include <sys/time.h>
#include <arpa/nameser.h>
#include <resolv.h>
#include <pthread.h>
#include "udt_priv.h"
#include <time.h>

#define MAX_MSG_SIZE	(10 * 1024)

static int exit_flag = 0;
int create_udtsock(int *udt_sock)
{
	udtp_startup();
	do{
		*udt_sock = udtp_socket(AF_INET, SOCK_DGRAM, 0);
		if(*udt_sock < 0){
			sleep(1);
			continue;
		}
		break;
	}while(!exit_flag);

	return 0;
}
#if 0
int set_udtp_socketopt(int udt_sock)
{
    int val, mss;
    struct linger udtlinger;
    
    /* udp mss 大小 */
    mss = 1052;
    udtp_setsockopt(udt_sock, 0,
            UDT_MSS, (char *)&mss, sizeof(int));
    /* udt发送buff大小 */
	mss -= 28;
    val = 160 * mss;
    udtp_setsockopt(udt_sock, 0, 
			UDT_SNDBUF, (char *)&val, sizeof(int));
    /* udt接收buff大小 */
    //val = 64 * mss;
    val = 200 * mss;
    udtp_setsockopt(udt_sock, 0,
            UDT_RCVBUF, (char *)&val, sizeof(int));
    /* udp 发送buff大小 */
    val = 8 << 10;
    udtp_setsockopt(udt_sock, 0,
            UDP_SNDBUF, (char *)&val, sizeof(int));
    /* udp 接收buff大小 */
    val = 16 << 10;
    udtp_setsockopt(udt_sock, 0,
            UDP_RCVBUF, (char *)&val, sizeof(int));
    /* 阻塞方式发送 */
    val = 1;
    udtp_setsockopt(udt_sock, 0,
            UDT_SNDSYN, (char *)&val, sizeof(int));
    /* 非阻塞方式接收 */
    val = 0;
    udtp_setsockopt(udt_sock, 0,
            UDT_RCVSYN, (char *)&val, sizeof(int));
    /* 发送超时值10秒 */
    val = 10000;
    udtp_setsockopt(udt_sock, 0,
			UDT_SNDTIMEO, (char *)&val, sizeof(int));
    /* socket close后一秒处理socket buffer，udt_close处理阻塞时间为2秒 
	 * 当用密码错误时，如果udt_close不阻塞，那错误密码信息不会发到用户那里*/
    udtlinger.l_onoff = 1;
    udtlinger.l_linger = 2;
    udtp_setsockopt(udt_sock, 0,
            UDT_LINGER, (char *)&udtlinger, sizeof(udtlinger));
}
#endif
int set_udtp_socketopt(int udt_sock)
{
    int val, mss;
//    struct linger udtlinger;
#if 0    
	val = 500;// mininum is 32, little than 32 is invalid
	udtp_setsockopt(udt_sock,0,UDT_FC,(char*)&val,sizeof(int));
#endif
    /* udp mss 大小 */
    mss = 1052;
    udtp_setsockopt(udt_sock, 0,
            UDT_MSS, (char *)&mss, sizeof(int));
	/*udt rcv buf*/
	mss -= 28;
	val = 300 * mss;
	udtp_setsockopt(udt_sock, 0, UDT_RCVBUF, (char *)&val, sizeof(int));
    /* udp 接收buff大小 */
    val = 32 << 10;
    udtp_setsockopt(udt_sock, 0,
            UDP_RCVBUF, (char *)&val, sizeof(int));
    /* udt发送buff大小 */
    val = 80 * 1052;
    udtp_setsockopt(udt_sock, 0, 
			UDT_SNDBUF, (char *)&val, sizeof(int));
    /* udp 发送buff大小 */
    val = 8 << 10;
    udtp_setsockopt(udt_sock, 0,
            UDP_SNDBUF, (char *)&val, sizeof(int));

    /* 阻塞方式发送 */
    val = 1;
    udtp_setsockopt(udt_sock, 0,
            UDT_SNDSYN, (char *)&val, sizeof(int));
    /* 发送超时值10秒 */
    val = 10000;
    udtp_setsockopt(udt_sock, 0,
			UDT_SNDTIMEO, (char *)&val, sizeof(int));
	return 0;
}
void *recv_data(void *arg)
{
	int sock = *((int*)arg);
	int rcv_len = 0;
	int ret;
	char rcv_buf[MAX_MSG_SIZE];
	struct timeval timeout;
	time_t time_bak,time_now;
	int rcv_total = 0;
	double speed = 0.0;

	printf("this is udt rcv thread sock:%d\n", sock);
	while(!exit_flag)
	{
		timeout.tv_sec = 5;
		timeout.tv_usec = 0;
		udtp_udzero(1);
		udtp_udset(sock, 1);

		time_bak = time(NULL);
		ret = udtp_select(sock, 1, 0, &timeout);
		if(ret < 0){
			printf("udt recv select error\n");
			break;
		}else if(0 == ret){
			// printf("udt recv select timeout\n");
		}else{
			if(udtp_udisset(sock, 1)){
				memset(rcv_buf,0,sizeof(rcv_buf));
				rcv_len = udtp_recvmsg(sock, rcv_buf, MAX_MSG_SIZE);
				if(rcv_len < 1){
					printf("udt recvmsg < 1[%d]:%s\n", rcv_len, udtp_geterromessage());
					break;
				}else{
					rcv_total += rcv_len;
					//printf("recv len:%d\n", rcv_len);
				}
			}
		}
		time_now = time(NULL);
		if(time_now-time_bak > 0)
		{// 1s calculate the recv speed, unit is Kbytes/s
			speed = (double)rcv_total/(time_now-time_bak);
			printf("recvmsg speed = %6.2lfkbps\n", speed*8/1024);
			time_bak = time_now;
			rcv_total = 0;
		}
		//usleep(300000);
	}
	udtp_close(sock);
	return 0;
}

void *tcp_server(void *arg)
{
	int ret = 0;
	int listenfd = -1;
	struct sockaddr_in server_addr;
	socklen_t socklen = sizeof(server_addr);
	
	memset(&server_addr, 0, socklen);

	listenfd = socket(AF_INET, SOCK_STREAM,0);
	server_addr.sin_family = AF_INET;
	server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
	server_addr.sin_port = htons(4900);

	int val = 1;
	setsockopt(listenfd,SOL_SOCKET,SO_REUSEADDR,&val,sizeof(val));
//	struct timeval rcvto;
//	rcvto.tv_sec = 5;
//	rcvto.tv_usec = 0;
//	setsockopt(listenfd,SO_RCVTIMEO, &rcvto,sizeof(rcvto));

	if(bind(listenfd, (struct sockaddr*)&server_addr, socklen)<0)
	{
		printf("bind error:%s\n",strerror(errno));
		return NULL;
	}
	if(listen(listenfd, 10) < 0)
	{
		printf("listen error:%s\n",strerror(errno));
		return NULL;
	}
	fd_set rdfds;
	fd_set tmpfds;
	int maxfd = listenfd;
	int clifd[16];
	char buff[1024]={0};
	int i,len = 1024;;
	FD_ZERO(&rdfds);
	FD_ZERO(&tmpfds);
	FD_SET(listenfd, &rdfds);
	for(i=0;i<16;i++)
		clifd[i]=-1;
	while(1)
	{
		tmpfds = rdfds;
		if(select(maxfd+1,&tmpfds,NULL,NULL,NULL)>0)
		{
			if(FD_ISSET(listenfd,&tmpfds))
			{
				for(i=0;i<16;i++)
				{
					if(clifd[i]<0)
						clifd[i] = accept(listenfd, NULL, NULL);
					if(clifd[i] < 0)
						break;
					FD_SET(clifd[i],&rdfds);
					if(clifd[i] > maxfd) maxfd = clifd[i];
					break;
				}
			}
			for(i=0;i<16;i++)
			{
				if(clifd[i]>0 && FD_ISSET(clifd[i],&tmpfds))
				{
					ret = recv(clifd[i], buff, len, 0);
					if(ret > 0)
					printf("recv from [%d],ret = %d\n",clifd[i],ret);
					else if(ret <= 0)
					{close(clifd[i]);FD_CLR(clifd[i],&rdfds);clifd[i]=-1;}
				}
			}
		}
	}
	return NULL;
}

int main(int argc, char **argv)
{
	int udt_sock = -1, client_sock = -1;
	int ret = 0, pth_count = 0;
	struct sockaddr_in udt_addr;
	struct timeval timeout;
	pthread_t pid[20];
	int udt_port = 12345;

	if(argc < 2)
	{
		fprintf(stderr,"Usage:%s <port> \n",argv[0]);
		exit(EXIT_FAILURE);
	}
	else
	{
		udt_port = atoi(argv[1]);
	}
	create_udtsock(&udt_sock);
	set_udtp_socketopt(udt_sock);

	memset(&udt_addr, 0, sizeof(struct sockaddr_in));
	udt_addr.sin_port = htons(udt_port);
	udt_addr.sin_family = AF_INET;
	udt_addr.sin_addr.s_addr = INADDR_ANY;
	if(P2P_UDT_ERRO == udtp_bind(udt_sock, (struct sockaddr*)&udt_addr, 
				sizeof(udt_addr))){
		printf("udt bind error\n");
		udtp_close(udt_sock);
		udt_sock = -1;
		return -1;
	}
	if(P2P_UDT_ERRO == udtp_listen(udt_sock, 10)){
		udtp_close(udt_sock);
		udt_sock = -1;
		return -1;
	}

	/* add tcp server thread */
//	pthread_t tcpid;
//	pthread_create(&tcpid,NULL, tcp_server,NULL);

	while(!exit_flag)
	{
		timeout.tv_sec = 5;
		timeout.tv_usec = 0;

		udtp_udzero(1);
		udtp_udset(udt_sock, 1);
		ret = udtp_select(udt_sock, 1, 0, &timeout);
		if(ret < 0){
			printf("udt server select error\n");
			sleep(1);
		}else if(0 == ret){
			usleep(30000);
		}else{
			if(udtp_udisset(udt_sock, 1)){
				client_sock = udtp_accept(udt_sock, NULL, NULL);
				if(P2P_UDT_ERRO == client_sock){
					continue;
				}else{
					printf("create udt recv data thread sock:%d\n", client_sock);
					pthread_create(&pid[pth_count], NULL, recv_data, &client_sock);
					//pthread_t rcv_pid;
					//pthread_create(&rcv_pid, NULL, recv_data, &client_sock);
					pth_count ++;
				}
			}
		}
	}
	udtp_close(udt_sock);

	return 0;
}

